package async;

import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: linKe
 * @Date: 2022/1/14 17:38
 * @FileName: ThreadAsyncExecutor
 * @version: 1.0
 * @Description: 描述
 */
public class ThreadAsyncExecutor implements AsyncExecutor{
    /**
     * 线程计数
     */
    private static final AtomicInteger COUNTER = new AtomicInteger(0);

    @Override
    public <T> AsyncResult<T> startProcess(Callable<T> task) {
        return startProcess(task, null);
    }

    @Override
    public <T> AsyncResult<T> startProcess(Callable<T> task, AsyncCallback<T> callback) {
        CompletableResult<T> asyncResult = new CompletableResult<T>(callback);
        //启动一个worker线程，此处使用线程池更好 匿名创建线程池
        new Thread(() -> {
            try {
                asyncResult.setValue(task.call());
            } catch (Exception e) {
                asyncResult.setException(e);
            }
        }, "worker-" + COUNTER.getAndIncrement()).start();
        return asyncResult;
    }

    @Override
    public <T> T endProcess(AsyncResult<T> result) throws InterruptedException, ExecutionException {
        if (!result.isComplete()) {
            result.await();
        }
        return result.getValue();
    }

    private static class CompletableResult<T> implements AsyncResult<T> {

        //运行状态
        private static final int RUNNING = 1;
        private static final int FAILED = 2;
        private static final int FINSHED = 3;

        private volatile int status = RUNNING;
        //最终的执行结果
        private T value;
        private Exception exception;
        private final Object lock;
        private final Optional<AsyncCallback<T>> callback;

        public CompletableResult(AsyncCallback<T> callback) {
            this.lock = new Object();
            this.callback = Optional.ofNullable(callback);
        }

        /**
         * worker线程执行完毕，调用此方法设置最终结果
         * @param value 值
         */
        public void setValue(T value) {
            this.status = FINSHED;
            this.value = value;
            callback.ifPresent(cb -> {
                cb.onComplete(value, Optional.empty());
            });
            //通知所有等待的线程
            synchronized (lock) {
                lock.notifyAll();
            }
        }

        /**
         * worker线程抛出异常
         * @param exception 异常
         */
        public void setException(Exception exception) {
            this.status = FAILED;
            this.exception = exception;
            callback.ifPresent(cb -> {
                cb.onComplete(value, Optional.of(exception));
            });
            synchronized (lock) {
                lock.notifyAll();
            }
        }

        @Override
        public boolean isComplete() {
            // 状态判断 
            return status > RUNNING;
        }


        @Override
        public T getValue() throws ExecutionException {
            // 完成状态
            if (status == FINSHED) {
                return value;
                // 失败
            } else if (status == FAILED) {
                throw new ExecutionException(exception);
            } else {
                throw new IllegalStateException("Execution not complete yet");
            }
        }

        /**
         * 等待锁
         * @throws InterruptedException 执行中断异常
         */
        @Override
        public void await() throws InterruptedException {
            synchronized (lock) {
                if (!isComplete()) {
                    lock.wait();
                }
            }
        }

    }
}
